package cn.tedu.micro.demo.rabbitmq;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.Map;

/**
 * 测试消费逻辑代码
 */
@Slf4j
public class MyConsumer {
    private Channel channel;
    /**
     * @Before注解 junit提供的功能 作用是在所有@Test方法执行之前 执行,一般是初始化功能
     */
    @Before
    public void createChannel(){
        try{
            //1.创建连接工厂-工厂模式 创建连接的过程 被工厂封装了
            ConnectionFactory factory = new ConnectionFactory();
            //2.给工厂提供连接创建的必要属性
            //2.1rabbitmq host
            factory.setHost("192.168.8.100");
            //2.2rabbitmq port
            factory.setPort(5672);
            //2.3账号权限
            factory.setUsername("admin");
            factory.setPassword("123456");
            //3.生成一个长连接
            Connection connection = factory.newConnection();
            //4.创建通道短链接
            channel = connection.createChannel();
        }catch (Exception e){
            log.error("创建连接失败");
        }
    }
    //消费逻辑
    @Test
    public void receiveMessage(){
        /**
         * String queue: 队列名称
         * boolean autoAck: 是否自动确认,只要拿到消息,立刻反馈给rabbitmq消费成功
         * DeliveryCallback delivery: 异步回调函数,消息达到之后的消费逻辑 交给他处理
         * CancleCallback callback: 异步回调函数,消费者取消消费 取消订阅时调用,这里用不到
         */
        //匿名内部类
        DeliverCallback delivery=new DeliverCallback(){
            /**
             * 当连接rabbitmq 监听消息队列 存在传递的消息时,handle
             * 消费逻辑执行者
             * String consumerTag: 调用回调消费者 表示 唯一值 识别是谁在消费消息
             * Delivery message: 消息封装对象 包含消息内容body 消息属性properties 消息头header
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //解析消息对象数据 反序列化body
                byte[] body = message.getBody();
                String msg=new String(body);
                log.info("消费者01拿到消息:{}",msg);
                //确认失败 requeue 失败之后是否允许消息 重新投递
                log.info("消费者01确认消息失败,重新投递");
                channel.basicReject(message.getEnvelope().getDeliveryTag(), true);
            }
        };
        CancelCallback cancel=new CancelCallback(){
            /**
             * 消费者取消消费的回调函数
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                log.error("消费者取消消费",consumerTag);
            }
        };
        try {
            channel.basicConsume("queue1",false,delivery,cancel);
        } catch (IOException e) {
            log.error("消费逻辑失败",e);
        }
        while(true);
    }
    @Test
    public void getAndRejectMessage(){
        try {
            channel.basicConsume(
                    "biz_queue",
                    false,
                    (consumerTag, message) -> {
                        //消费实际执行方法 手动确认拒绝
                        log.info("消费演示拒绝");
                        channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
                    },
                    consumerTag -> log.error("消费者取消消费",consumerTag));
        } catch (IOException e) {
            log.error("消费逻辑失败",e);
        }
        while(true);
    }
}
